package com.farmer.reliablemessage.core;

import com.farmer.reliablemessage.CommonConfiguration;
import com.farmer.reliablemessage.ReliableMessageProperties;
import com.farmer.reliablemessage.api.IDApi;
import com.farmer.reliablemessage.entity.ReliableMessage;
import com.farmer.reliablemessage.event.ReliableMessageSendEvent;
import com.farmer.reliablemessage.mapper.MessageMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionalEventListener;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.DigestUtils;

import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author tech-famer
 * @ClassName: MQTemplate
 * @Description: (这里用一句话描述这个类的作用)
 * @date 2021/11/1 16:30
 */
@Component
public class MQTemplate {

    private Logger logger = LoggerFactory.getLogger(MQTemplate.class);

    private final IDApi idApi;
    @Value("${spring.application.name}")
    private String appName;
    private MessageMapper messageMapper;
    private ApplicationEventPublisher applicationEventPublisher;
    private ConcurrentHashMap<String, HashSet<MessageWrapper>> messageMap = new ConcurrentHashMap<>();

    private AtomicInteger count = new AtomicInteger(0);

    private ReliableMessageProperties properties;


    public MQTemplate(ReliableMessageProperties properties, MessageMapper messageMapper, IDApi idApi, ApplicationEventPublisher applicationEventPublisher) {
        this.messageMapper = messageMapper;
        this.idApi = idApi;
        this.applicationEventPublisher = applicationEventPublisher;
        this.properties = properties;
        logger.info("MQTemplate load success!");
    }

    public void saveAndSendMessage(final String exchange, final String routingkey, final String object) {
        final ReliableMessage message = new ReliableMessage();
        message.setMsgid(String.valueOf(this.idApi.nextID()));
        message.setPayload(object);
        message.setOwner(CommonConfiguration.getLockID());
        message.setSource(this.appName);
        message.setExchange(exchange);
        message.setRoutingkey(routingkey);
        this.messageMapper.saveMessage(message);

        //获取当前事务名称，无事务则直接发送消息，有事务则事务提交后发送消息
        String currentTransactionName = TransactionSynchronizationManager.getCurrentTransactionName();
        if(currentTransactionName == null){
            this.applicationEventPublisher.publishEvent(new ReliableMessageSendEvent(this, message));
        }else{
            String key = DigestUtils.md5DigestAsHex((Thread.currentThread().getName() + currentTransactionName).getBytes());
            logger.info(key);
            synchronized (MQTemplate.class){
                if(messageMap.containsKey(key)){
                    messageMap.get(key).add(new MessageWrapper(message));
                }else{
                    messageMap.put(key,new HashSet<MessageWrapper>(){{add(new MessageWrapper(message));}});
                    applicationEventPublisher.publishEvent(key);
                }
            }
        }
    }



    @TransactionalEventListener
    public void handleTransactionalSubmit(Object object){
        if(object instanceof String){
            HashSet<MessageWrapper> messages = messageMap.get(object);
            if(messages!= null){
                for (MessageWrapper messageWrapper : messages) {
                    this.applicationEventPublisher.publishEvent(new ReliableMessageSendEvent(this, messageWrapper.getMessage()));
                }
            }
            messageMap.remove(object);
        }

        if (count.incrementAndGet() >= properties.getRemoveLimit()) {
            count.set(0);
            try {
                long l = System.currentTimeMillis();
                for (Map.Entry<String, HashSet<MessageWrapper>> stringHashSetEntry : messageMap.entrySet()) {
                    for (MessageWrapper messageWrapper : stringHashSetEntry.getValue()) {
                        if(l - messageWrapper.getIntime() >= properties.getExpireTime()){
                            messageMap.remove(stringHashSetEntry.getKey());
                            break;
                        }
                    }
                }
            }catch (Exception e){
                logger.warn("schedule remove expire reliablemessage error");
            }
        }

    }
}
